using System;
using System.Collections;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using Me.Prettyprint.Cassandra.Connection.Factory;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;

namespace Me.Prettyprint.Cassandra.Connection
{
    /// <summary>
    /// The current implementation for Dynamic load balancing uses the following settings.
    /// 
    /// update interval - 100
    /// reset interval - 20000
    /// dynamic badness threshold - 0.10
    /// 
    /// When HConnectionManager asks for a connection from the pool, the class 
    /// scores and sorts the connection pool before return a connection.
    /// </summary>
	public class DynamicLoadBalancingPolicy : LoadBalancingPolicy
	{
        private static readonly Logger logger = new Logger(typeof(DynamicLoadBalancingPolicy));
        private ConcurrentDictionary<HClientPool, double> scores = new ConcurrentDictionary<HClientPool,double>();
        private ConcurrentBag<LatencyAwareHClientPool> allPools = new ConcurrentBag<LatencyAwareHClientPool>();
		
		// default values this can be changed by the Client.
		private int UPDATE_INTERVAL = 100;
		private int RESET_INTERVAL = 20000;
		private double DYNAMIC_BADNESS_THRESHOLD = 0.10;
        private CancellationTokenSource updateTokenSource = new CancellationTokenSource();
        private CancellationTokenSource resetTokenSource = new CancellationTokenSource();
        private Task updateScoresTask;
        private Task resetIntervalTask;
		
		public DynamicLoadBalancingPolicy()
		{
            var utsource = updateTokenSource.Token;
            var rtsource = resetTokenSource.Token;
            updateScoresTask = Task.Factory.StartNew(() => UpdateScores());
            resetIntervalTask = Task.Factory.StartNew(() => ClearAllPools());
		}
		public int UpdateInterval
		{
			get
			{
				return UPDATE_INTERVAL;
			}
			set
			{
				UPDATE_INTERVAL = value;
			}
		}
		public int ResetInterval
		{
			get
			{
				return RESET_INTERVAL;
			}
			set
			{
				RESET_INTERVAL = value;
			}
		}
		public double BadnessThreshold
		{
			get
			{
				return DYNAMIC_BADNESS_THRESHOLD;
			}
			set
			{
				DYNAMIC_BADNESS_THRESHOLD = value;
			}
		}
        public HClientPool CreateConnection(HClientFactory clientFactory, CassandraHost host, CassandraClientMonitor monitor)
		{
			LatencyAwareHClientPool pool = new LatencyAwareHClientPool(clientFactory, host, monitor);
			Add(pool);
			return pool;
		}
        public HClientPool GetPool(ICollection<HClientPool> pools, List<CassandraHost> excludeHosts)
        {
            Dictionary<CassandraHost, HClientPool> pooldictionary = new Dictionary<CassandraHost, HClientPool>();
            foreach (HClientPool p in pools)
            {
                pooldictionary.Add(p.CassandraHost, p);
            }
            if (excludeHosts != null)
            {
                foreach (CassandraHost host in excludeHosts)
                {
                    if (pooldictionary.ContainsKey(host))
                    {
                        pooldictionary.Remove(host);
                    }
                }
            }
            SortByScoreComparator comparator = new SortByScoreComparator(this);
            List<HClientPool> sublist = new List<HClientPool>(pooldictionary.Values);
            sublist.Shuffle();
            HClientPool hp = sublist[0];
            double first = scores[hp];
            for (int i = 1; i < sublist.Count; i++)
            {
                HClientPool nextpool = sublist[i];
                double nextd = scores[nextpool];
                if ((first - nextd) / first > DYNAMIC_BADNESS_THRESHOLD)
                {
                    sublist.Sort(comparator);
                    break;
                }
            }
            HClientPool pool = sublist[0];
            logger.Info(" -------- " + pool.CassandraHost.Ip + " -------------");
            return pool;
        }
		internal void Add(LatencyAwareHClientPool pool)
		{
			allPools.Add(pool);
			// update the reference of the scores intially it is Zero.
			scores.AddOrUpdate(pool, 0.0, (key,oldvalue) => 0.0);
		}
		internal void UpdateScores()
		{
			foreach (LatencyAwareHClientPool pool in this.allPools)
			{
				scores.TryAdd(pool, pool.Score());
				pool.ResetIntervel();
			}
		}
        internal void ClearAllPools()
        {
            foreach (LatencyAwareHClientPool p in allPools)
            {
                p.Clear();
            }
        }
		private class SortByScoreComparator : System.Collections.Generic.IComparer<HClientPool>
		{
			public SortByScoreComparator(DynamicLoadBalancingPolicy enclosingInstance)
			{
                LoadBalancingPolicy = enclosingInstance;
			}
			public DynamicLoadBalancingPolicy LoadBalancingPolicy
			{
				get;
				private set;
			}
			public virtual int Compare(HClientPool p1, HClientPool p2)
			{
                double scored1 = 0;
                LoadBalancingPolicy.scores.TryGetValue(p1, out scored1);
                double scored2 = 0;
                LoadBalancingPolicy.scores.TryGetValue(p2, out scored2);
				if (scored1.Equals(scored2))
					return 0;
				if (scored1 < scored2)
					return - 1;
				else
					return 1;
			}
			virtual public System.Int32 Compare(System.Object x, System.Object y)
			{
				return 0;
			}
		}
	}
}